package com.ds.infrastructure.hyperspace.container.endpoint.connector.health;

import com.ds.infrastructure.hyperspace.common.EndpointProtocolEnum;
import com.ds.infrastructure.hyperspace.common.config.TargetNodeConfig;
import com.ds.infrastructure.hyperspace.common.config.TargetProtocolConfig;
import com.ds.infrastructure.hyperspace.container.endpoint.destination.TargetEndpointNode;
import com.ds.infrastructure.hyperspace.container.internal.http2.HyperspaceHttp2Codec;
import io.netty.channel.Channel;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.handler.codec.http.HttpClientCodec;
import io.netty.handler.codec.http.HttpObjectAggregator;
import lombok.extern.slf4j.Slf4j;

import static com.ds.infrastructure.hyperspace.common.EndpointProtocolEnum.HTTP2;
import static com.ds.infrastructure.hyperspace.common.EndpointProtocolEnum.HTTP2C;

/**
 * @author  Weishaoying
 */
@Slf4j
public class HealthCheckChannelInitializer extends ChannelInitializer<Channel> {

    private final TargetEndpointNode targetEndpointNode;

    private final HyperspaceNodeHealthChecker nodeHealthChecker;

    public HealthCheckChannelInitializer(TargetEndpointNode targetEndpointNode,
            HyperspaceNodeHealthChecker nodeHealthChecker) {
        this.targetEndpointNode = targetEndpointNode;
        this.nodeHealthChecker = nodeHealthChecker;
    }

    @Override
    protected void initChannel(Channel ch) throws Exception {
        ChannelPipeline pipeline = ch.pipeline();
        TargetNodeConfig nodeConfig = targetEndpointNode.getNodeConfig();
        TargetProtocolConfig protocolConfig = nodeConfig.getProtocolConfig();
        EndpointProtocolEnum protocol = nodeConfig.getProtocol();
        if (protocol.getIndex() < HTTP2.getIndex()) {
            pipeline.addLast(new HttpClientCodec(
                    protocolConfig.getMaxInitialLineLength(),
                    protocolConfig.getMaxHeaderSize(),
                    protocolConfig.getMaxChunkSize(),
                    false,
                    false,
                    protocolConfig.getInitialBufferSize()));
            pipeline.addLast(new HttpObjectAggregator(protocolConfig.getMaxBodySize()));
            pipeline.addLast(new Http1HealthCheckHandler(targetEndpointNode, nodeHealthChecker));
            return;
        } else if (protocol.getIndex() <= HTTP2C.getIndex()) {
            HyperspaceHttp2Codec hyperspaceHttp2Codec = new HyperspaceHttp2Codec(false, protocolConfig);
            pipeline.addLast(hyperspaceHttp2Codec);
            pipeline.addLast(new Http2HealthCheckHandler(hyperspaceHttp2Codec,
                    targetEndpointNode, nodeHealthChecker));
            return;
        }
        log.error("Health check for protocol {} is not support!", protocol);
        ch.close();
    }
}
